package com.mmc.delay.timer;

import com.mmc.delay.api.DelayFacade;
import com.mmc.delay.api.JobHandle;
import com.mmc.delay.bean.Job;
import com.mmc.delay.config.DelayCoreConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class ReadyQueueTimer {

    private DelayFacade delayFacade;

    private ReadyQueueTimer(){}

    public ReadyQueueTimer(DelayFacade delayFacade) {
        this.delayFacade = delayFacade;
    }

    private volatile boolean consumeStop = false;

    private static final Logger logger = LoggerFactory.getLogger(ReadyQueueTimer.class);


    public static final ThreadPoolExecutor JOB_HANDLE_THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(4, 10,
            60L, TimeUnit.SECONDS,
            new LinkedBlockingDeque<Runnable>(3000));

    public static final ThreadPoolExecutor READY_CONSUME_THREAD_POOL_EXECUTOR=new ThreadPoolExecutor(1,1,0,TimeUnit.SECONDS,new LinkedBlockingDeque<Runnable>(100), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread=new Thread(r);
            thread.setName("ready-consume-thread");
            return thread;
        }
    });

    public void runConsume(){
        READY_CONSUME_THREAD_POOL_EXECUTOR.submit(new Runnable() {
            @Override
            public void run() {
                while (!consumeStop){
                    doRun();
                }
            }
        });

    }

    private void doRun() {
        List<Job> list = new ArrayList<>();
       try {
          list = delayFacade.popReadyJob();
        } catch (Exception e) {
            logger.error("ready list取出任务异常", e);
        }
        for (final Job job : list) {
            boolean result = false;
            FutureTask<Boolean> futureTask = null;
            try {
                futureTask=new FutureTask<>(new Callable<Boolean>() {
                    @Override
                    public Boolean call() throws Exception {
                        JobHandle jobHandle = DelayCoreConfig.getJobHandle(job.getTopic());
                        return jobHandle.handle(job);
                    }
                });
                JOB_HANDLE_THREAD_POOL_EXECUTOR.submit(futureTask);
                result=futureTask.get(DelayCoreConfig.getTopicConfig(job.getTopic()).getTimeOut(),TimeUnit.MILLISECONDS);
            }catch (InterruptedException | ExecutionException| TimeoutException e) {
                logger.error("任务执行过程中超时或中断,任务id:{}", job.getJobId(), e);
                futureTask.cancel(true);
            }
            catch (Exception e) {
                logger.error("任务执行过程中异常,任务id:{}", job.getJobId(), e);
            }
            try {
                if (result) {
                    delayFacade.removeJob(job.getJobId());
                } else {
                    delayFacade.tryRetry(job);
                }
            } catch (Exception e) {
                logger.error("任务移除异常.jobId:{}",job.getJobId(), e);
            }
        }

    }

    public void stopConsume(){
        this.consumeStop =true;
    }


}
